package socket.selectorThread;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class SelectorThread implements Runnable {
  /**
   * 每个线程 对应一个 Selector
   * 多线程的情况下，该主机，该程序的并发客户端被分配到多个selector上
   * 注意，每个客户端 只能绑定一个selector
   * 多个线程之间不会有交互问题
   */
  
  Selector selector;
  SelectorThreadGroup selectorThreadGroup;
  LinkedBlockingQueue<Channel> queue = new LinkedBlockingQueue<>();
  
  SelectorThreadGroup stg;
  
  public SelectorThread(SelectorThreadGroup selectorThreadGroup) throws IOException {
    this.selectorThreadGroup = selectorThreadGroup;
    // open 其实是一个 poll_create 的过程
    this.selector = Selector.open();
  }
  
  @SuppressWarnings("InfiniteLoopStatement")
  @Override
  public void run() {
    // 利用 一个子线程来执行 这个loop
    while (true) {
      try {
        // 这里是 会进行阻塞的
        int descriptorCount = selector.select();
        
        // 如果 描述符 > 0的话 开始执行处理
        if (descriptorCount > 0) {
          // 这里能 拿到 描述符 有可能是读/ 写/ 待请求
          Set<SelectionKey> selectionKeys = selector.selectedKeys();
          Iterator<SelectionKey> iterator = selectionKeys.iterator();
          
          while (iterator.hasNext()) {
            SelectionKey key = iterator.next();
            iterator.remove();
            
            if (key.isAcceptable()) {
              acceptHandler(key);
            } else if (key.isReadable()) {
              readHandler(key);
            } else if (key.isWritable()) {
              System.out.println("这是一个 write 处理");
            } else {
              System.out.println("else ......");
            }
          }
        }
        
        // 如果执行到这里，说明 文件描述符为空的
        // 这里要 处理一些 task 任务
        if (!queue.isEmpty()) {
          Channel channel = queue.take();
          // 可能是  server socket
          if (channel instanceof ServerSocketChannel) {
            ServerSocketChannel server = (ServerSocketChannel) channel;
            server.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println(Thread.currentThread().getName() + " register server listen");
            
            // 可能是 client socket
          } else if (channel instanceof SocketChannel) {
            SocketChannel client = (SocketChannel) channel;
            ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
            client.register(selector, SelectionKey.OP_READ, buffer);
            System.out.println(Thread.currentThread().getName() + " register client: " + client.getRemoteAddress());
          }
        }
      } catch (IOException | InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
  
  /**
   * 等待接受的方法
   *
   * @param key Selector key
   */
  private void acceptHandler(SelectionKey key) throws IOException {
    System.out.println(Thread.currentThread().getName() + ":accept ......");
    
    ServerSocketChannel server = (ServerSocketChannel) key.channel();
    SocketChannel client = server.accept();
    // 修改为 非阻塞模式
    client.configureBlocking(false);
    
    // 将 client 绑定到某个 Selector
    // 这里 已轮训的方式 将不同的client 绑定到 不同的 Selector
//    this.selectorThreadGroup.nextSelector(client);
    this.selectorThreadGroup.nextSelectorV3(client);
  }
  
  /**
   * 这是一个读操作
   *
   * @param key Selector key
   */
  private void readHandler(SelectionKey key) throws IOException {
    System.out.println(Thread.currentThread().getName() + ":read ......");
    
    ByteBuffer buffer = (ByteBuffer) key.attachment();
    SocketChannel client = (SocketChannel) key.channel();
    
    buffer.clear();
    
    while (true) {
      int num = client.read(buffer);
      if (num > 0) {
        // 直接将内容翻转
        buffer.flip();
        while (buffer.hasRemaining()) {
          client.write(buffer);
        }
        buffer.clear();
      } else if (num == 0) break;
      else {
        System.out.println("client: " + client.getRemoteAddress() + " closed......");
        key.cancel();
        break;
      }
    }
  }
  
  public void setWorker(SelectorThreadGroup stg) {
    this.stg = stg;
  }
}
